package com.gugee.ins.data.task.statistics.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.common.task.model.TaskRecord;
import com.gugee.common.task.service.AbstractProducerHandler;
import com.gugee.ins.common.dal.service.statistics.InsStBloggerCfsAvgDao;
import com.gugee.ins.data.common.msg.InsMBIdenticalLevelMessage;
import com.gugee.ins.data.task.config.TaskConfig;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
@Slf4j
@Service
public class StBloggerIdenticalLevelSvc extends AbstractProducerHandler<InsMBIdenticalLevelMessage> {

    @Autowired
    InsStBloggerCfsAvgDao insStBloggerCfsAvgDao;

    @Autowired
    @Qualifier("insBloggerIdenticalLevelMsgBaseInfoProducer")
    KafkaTemplate<Long, InsMBIdenticalLevelMessage> insBloggerIdenticalLevelMsgBaseInfoProducer;


    @Autowired
    public TaskConfig taskConfig;

    @Override
    public List<InsMBIdenticalLevelMessage> findMessageObject(TaskRecord taskRecord, TaskParam taskParam) {
        taskRecord.setTaskDesc("大播主同等级平均值统计");
        List<InsMBIdenticalLevelMessage> rst = insStBloggerCfsAvgDao.getInsMBIdenticalLevelMessageList(0,1000);
        return rst==null?new ArrayList<>():rst;
    }

    @Override
    public long objectTotals() {
        return 1000;
    }

    @Override
    public void sendMsg(List<InsMBIdenticalLevelMessage> messages, String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(taskConfig.getRetryBloggerTimes());
            msg.setRetryLimitTimes(taskConfig.getRetryBloggerTimes());
            insBloggerIdenticalLevelMsgBaseInfoProducer.send(topic,msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,messages.size());
        log.info("投递主题={},投递消息数={}", topic,messages.size());
    }

    @Override
    public void sendMsg2Kafka(int taskType, String who, TaskParam taskParam){
        long start=System.currentTimeMillis();
        //获取将发送的位置
        TaskRecord taskRecord =getTaskRecord(taskType,who);
        String currentTime= LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        if(taskRecord.getCurrentPage()==0){
            taskRecord.setFirstPageStartTime(currentTime);
            taskRecord.setFirstPageStartTimeMills(System.currentTimeMillis());
            taskRecord.setRecords(0L);
        }

        List<InsMBIdenticalLevelMessage> messages =findMessageObject(taskRecord, taskParam);

        if(messages==null){
            messages =new ArrayList<>(0);
        }

        //记录此次的发送情况
        log.info(" who={}, size={},page={}, pageSize={},  objectId from  {} to {},times={}秒",who,messages.size(),taskRecord.getCurrentPage(),taskParam.getRecords(),taskRecord.getPreStartObjectId(),taskRecord.getPreEndObjectId(),(System.currentTimeMillis()-start)/1000);
        XxlJobLogger.log("  size={},page={}, pageSize={},  record from  {} to {},times={}秒",messages.size(),taskRecord.getCurrentPage(),taskParam.getRecords(),taskRecord.getPreStartObjectId(),taskRecord.getPreEndObjectId(),(System.currentTimeMillis()-start)/1000);


        sendMsg(messages,taskParam.getTopic());
        taskRecord.setRecords(taskRecord.getRecords() + messages.size());
        taskRecord.setLastRecords((long)messages.size());
        //记录累积发送次数
        taskRecord.setCurrentPage(taskRecord.getCurrentPage()+1);
        taskRecord.setUpdateTime(new Date());
        taskRecordSvc.save(taskRecord);

    }
}
